package com.yuma.tingshu.search.service.impl;

import co.elastic.clients.elasticsearch.ElasticsearchClient;
import co.elastic.clients.elasticsearch._types.FieldValue;
import co.elastic.clients.elasticsearch._types.SortOrder;
import co.elastic.clients.elasticsearch._types.aggregations.Aggregate;
import co.elastic.clients.elasticsearch._types.aggregations.LongTermsBucket;
import co.elastic.clients.elasticsearch._types.query_dsl.BoolQuery;
import co.elastic.clients.elasticsearch.core.SearchRequest;
import co.elastic.clients.elasticsearch.core.SearchResponse;
import co.elastic.clients.elasticsearch.core.search.CompletionSuggestOption;
import co.elastic.clients.elasticsearch.core.search.Hit;
import co.elastic.clients.elasticsearch.core.search.Suggestion;
import co.elastic.clients.json.JsonData;
import com.google.common.collect.Lists;
import com.yuma.tingshu.album.client.AlbumInfoFeignClient;
import com.yuma.tingshu.album.client.CategoryFeignClient;
import com.yuma.tingshu.common.constant.RedisConstant;
import com.yuma.tingshu.common.result.Result;
import com.yuma.tingshu.model.album.BaseCategory3;
import com.yuma.tingshu.model.base.BaseEntity;
import com.yuma.tingshu.model.search.AlbumInfoIndex;
import com.yuma.tingshu.model.search.SuggestIndex;
import com.yuma.tingshu.query.search.AlbumIndexQuery;
import com.yuma.tingshu.search.service.SearchService;
import com.yuma.tingshu.vo.album.AlbumStatVo;
import com.yuma.tingshu.vo.search.AlbumInfoIndexVo;
import com.yuma.tingshu.vo.search.AlbumSearchResponseVo;
import jakarta.annotation.Resource;
import lombok.AllArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.elasticsearch.client.elc.ElasticsearchTemplate;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Service;
import org.springframework.util.Assert;
import org.springframework.util.CollectionUtils;


import java.io.IOException;
import java.text.SimpleDateFormat;
import java.time.LocalDateTime;
import java.util.*;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.stream.Collectors;

@Slf4j
@Service
@SuppressWarnings({"unchecked"})
public class SearchServiceImpl implements SearchService {

    @Autowired
    private CategoryFeignClient categoryFeignClient;

    @Autowired
    private ElasticsearchClient elasticsearchClient;

    @Autowired
    private AlbumInfoFeignClient albumInfoFeignClient;

    @Autowired
    private RedisTemplate redisTemplate;

    @Autowired
    private ElasticsearchTemplate elasticsearchTemplate;

    @Resource(name = "commonThreadPoolConfig")
    private ExecutorService executorService;

    /**
     * GET /albuminfo/_search
     * {
     *   "size": 0,
     *   "query": {
     *     "terms": {
     *       "category3Id": [
     *         1001,
     *         1002,
     *         1003,
     *         1004,
     *         1005,
     *         1006,
     *         1007
     *       ]
     *     }
     *   },
     *   "aggs": {
     *     "cateAlbumGroup": {
     *       "terms": {
     *         "field": "category3Id"
     *       },
     *       "aggs": {
     *         "hotscoreTop": {
     *           "top_hits": {
     *             "size": 6,
     *             "sort": [
     *               {
     *                 "hotScore": {
     *                   "order": "desc"
     *                 }
     *               }
     *             ]
     *           }
     *         }
     *       }
     *     }
     *   }
     * }
     */
    @Override
    public List<Map<String, Object>> channel(Long category1Id) throws IOException {

        //1.创建返回类型数组充当查询后存放容器
        List<Map<String, Object>> maps = new ArrayList<>();

        // 2.调用远程接口：根据一级分类id查询置顶的三级分类,由于运程调用接口，返回值类型为Result<>,后续通过Result.getData()获取数据
        Result<List<BaseCategory3>> topBaseCategory3Result = categoryFeignClient.findTopBaseCategory3(category1Id);

        /// Assert.notNull 将 “判断空值 + 抛异常” 封装为一行代码，直接替代 3 行以上的手动逻辑，让代码更聚焦于核心功能
        Assert.notNull(topBaseCategory3Result, "首页获取置顶频道时，远程调用失败" + category1Id);

        List<BaseCategory3> baseCategory3List = topBaseCategory3Result.getData();

        // 判断三级分类是否为空,如果为空则返回空集合
        if (CollectionUtils.isEmpty(baseCategory3List)) {
            return maps;
        }
        // 3.从三级分类集合中获取三级分类id集合
        /*List<FieldValue> category3IdList = new ArrayList<>();
        baseCategory3List.forEach(baseCategory3 -> {
            category3IdList.add(FieldValue.of(baseCategory3.getId()));
        });*/
        List<FieldValue> category3IdList = baseCategory3List.stream().map(baseCategory3 ->
                FieldValue.of(baseCategory3.getId())
        ).toList();


        // 4.把三级分类集合处理成map<category3Id,BaseCategory3>
        /**
         * Collectors.toMap(  // 用Collectors工具类的toMap方法生成Map
         *             BaseEntity::getId,  // Map的“键（key）”：取每个分类实体的ID
         *             baseCategory3 -> baseCategory3  // Map的“值（value）”：取分类实体本身
         *         )
         */
        Map<Long, BaseCategory3> category3Map = baseCategory3List.stream().collect(Collectors.toMap(BaseEntity::getId, baseCategory3 -> baseCategory3));

        // 5.构建dsl,根据指定的①多个三级分类 ID，查询②每个分类下③热度最高的前 6 条专辑信息
        SearchRequest request = SearchRequest.of(s -> s
                .index("albuminfo")
                .size(0)
                .query(q -> q
                        .terms(t -> t
                                .field("category3Id")
                                .terms(ts -> ts
                                        .value(category3IdList))))
                .aggregations("categoryAgg", agg -> agg
                        .terms(t -> t
                                .field("category3Id"))
                        .aggregations("hotscoreTop", sagg -> sagg
                                .topHits(t -> t
                                        .size(6)
                                        .sort(st -> st
                                                .field(f -> f
                                                        .field("hotScore").order(SortOrder.Desc)
                                                )
                                        )
                                )
                        )
                )
        );
        System.out.println(request);

        // 6.执行搜索，获取搜索结果集
        SearchResponse<AlbumInfoIndex> response = elasticsearchClient.search(request, AlbumInfoIndex.class);

        // 解析搜索结果集，见response结构图
        Map<String, Aggregate> aggregations = response.aggregations();
        if (CollectionUtils.isEmpty(aggregations)) {
            return maps;
        }
        Aggregate aggregate = aggregations.get("categoryAgg");
        if (aggregate == null) {
            return maps;
        }
        // 7.获取三级分类聚合中的桶集合
        List<LongTermsBucket> buckets = aggregate.lterms().buckets().array();

        // 8.把每一个桶转化成一个map结构
        buckets.forEach(bucket -> {
            // 创建map
            Map<String, Object> map = new HashMap<>();

            // 向map中添加三级分类对象
            map.put("baseCategory3", category3Map.get(bucket.key()));

            // 向map中添加三级分类板块对应的热门专辑列表
            // 获取桶中的子聚合
            Aggregate subAgg = bucket.aggregations().get("hotscoreTop");
            List<Hit<JsonData>> hits = subAgg.topHits().hits().hits();

            // 把hits集合转化成albumInfoIndex集合
            /**
             * hit：Stream 流中当前遍历的元素，类型是 Hit<JsonData>（ES 的单个命中文档）
             * hit.source()：获取该命中文档的 原始 JSON 数据，返回类型是 JsonData（Spring Data Elasticsearch 中封装 ES JSON 数据的工具类，可理解为 “JSON 容器”）
             * to(AlbumInfoIndex.class)：JsonData 类的核心方法，作用是 将底层 JSON 数据自动反序列化为指定的 Java 实体类（这里是 AlbumInfoIndex）
             * 原理：Spring Data Elasticsearch 会根据 AlbumInfoIndex 类的字段名（与 ES 索引的字段名对应，如 albumId、hotScore），自动将 JSON 中的键值对映射到实体类的属性中，无需手动解析 JSON
             */
            map.put("list", hits.stream().map(hit -> hit.source().to(AlbumInfoIndex.class)));

            /**
             * 最终，每个分类对应的 Map 结构会是这样：
             * {
             *   "baseCategory3": { "id": 103, "name": "悬疑推理", "parentId": 20 }, // 分类完整信息
             *   "list": [ // 该分类下的热门专辑列表
             *     { "id": 1001, "name": "白夜行", "hotScore": 98.5, "announcerName": "张三" },
             *     { "id": 1002, "name": "盗墓笔记", "hotScore": 96.2, "announcerName": "李四" }
             *   ]
             * }
             */
            maps.add(map);
        });
        return maps;
    }

    @Override
    public AlbumSearchResponseVo search(AlbumIndexQuery query) {
        try {
            SearchResponse<AlbumInfoIndex> response = elasticsearchClient.search(this.buildDsl(query), AlbumInfoIndex.class);
            AlbumSearchResponseVo responseVo = this.parseResult(response);
            // 分页参数从请求参数中获取
            responseVo.setPageNo(query.getPageNo());
            responseVo.setPageSize(query.getPageSize());
            // 计算总页数
            long totalPage = responseVo.getTotal() / responseVo.getPageSize();
            responseVo.setTotalPages(responseVo.getTotal() % responseVo.getPageSize() == 0 ? totalPage : totalPage + 1);
            return responseVo;
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    /**
     *  GET /albuminfo/_search
     * {
     *   "query": {
     *     "bool": {
     *       "must": [
     *         {
     *           "multi_match": {
     *             "query": "音乐",
     *             "fields": [
     *               "albumTitle",
     *               "albumIntro",
     *               "announcerName"
     *             ]
     *           }
     *         }
     *       ],
     *       "filter": [
     *         {
     *           "term": {
     *             "category1Id": "2"
     *           }
     *         },
     *         {
     *           "term": {
     *             "category2Id": "104"
     *           }
     *         },
     *         {
     *           "term": {
     *             "category3Id": "1019"
     *           }
     *         },
     *         {
     *           "nested": {
     *             "path": "attributeValueIndexList",
     *             "query": {
     *               "bool": {
     *                 "must": [
     *                   {
     *                     "term": {
     *                       "attributeValueIndexList.attributeId": {
     *                         "value": 1
     *                       }
     *                     }
     *                   },
     *                   {
     *                     "term": {
     *                       "attributeValueIndexList.valueId": {
     *                         "value": 2
     *                       }
     *                     }
     *                   }
     *                 ]
     *               }
     *             }
     *           }
     *         }
     *       ]
     *     }
     *   },
     *   "sort": [
     *     {
     *       "hotScore": {
     *         "order": "desc"
     *       }
     *     }
     *   ],
     *   "from": 0,
     *   "size": 10,
     *   "highlight": {
     *     "fields": {
     *       "albumTitle": {
     *         "pre_tags": "<b color='red'>",
     *         "post_tags": "</b>"
     *       }
     *     },
     *     "pre_tags": "<font color='red'>",
     *     "post_tags": "</font>"
     *   },
     *   "_source": {
     *     "excludes": [
     *       "hotScore",
     *       "attributeValueIndexList"
     *     ]
     *   }
     * }
     */
    private SearchRequest buildDsl(AlbumIndexQuery query) {
        /**
         * 在分类路径为 2→104→1019 的专辑中,
         * 筛选出属性 1 的值为 2,
         * 且标题 / 简介 / 播音员名称包含"音乐"的内容,
         * 按热度从高到低取前 10 条,
         * 同时高亮显示标题中的"音乐"关键词,
         * 并隐藏不必要的字段
         */
        SearchRequest.Builder request = new SearchRequest.Builder();
        request.index("albuminfo");
        //must
        BoolQuery.Builder boolQuery = new BoolQuery.Builder();
        String keyword = query.getKeyword();
        if (!StringUtils.isEmpty(keyword)) {
            boolQuery.must(q -> q.multiMatch(mq -> mq.query(keyword).fields("albumTitle", "albumIntro", "announcerName")));
        }

        //分类
        Long category1Id = query.getCategory1Id();
        if (category1Id != null) {
            boolQuery.filter(q -> q.term(tq -> tq.field("category1Id").value(category1Id)));
        }
        Long category2Id = query.getCategory2Id();
        if (category2Id != null) {
            boolQuery.filter(q -> q.term(tq -> tq.field("category2Id").value(category2Id)));
        }
        Long category3Id = query.getCategory3Id();
        if (category3Id != null) {
            boolQuery.filter(q -> q.term(tq -> tq.field("category3Id").value(category3Id)));
        }

        //nested
        List<String> attributeList = query.getAttributeList();
        if(!CollectionUtils.isEmpty(attributeList)){
            attributeList.forEach(attr -> {
                String[] attrInfo = attr.split(":");
                if (attrInfo != null && attrInfo.length == 2 && StringUtils.isNumeric(attrInfo[0]) && StringUtils.isNumeric(attrInfo[1])) {
                    boolQuery.filter(q -> q
                            .nested(n -> n
                            .path("attributeValueIndexList")
                            .query(q1 -> q1
                                    .bool(b -> b
                                            .must(m -> m
                                                    .term(t -> t
                                                            .field("attributeValueIndexList.attributeId").value(attrInfo[0])))
                                            .must(m -> m
                                                    .term(t -> t
                                                            .field("attributeValueIndexList.valueId").value(attrInfo[1])))
                                    )
                            )
                    ));
                }
            });
        }


        //排序
        String order = query.getOrder();
        if (!StringUtils.isEmpty(order)) {
            String[] sorts = order.split(":");
            if (sorts != null && sorts.length == 2 && StringUtils.isNumeric(sorts[0]) && List.of("desc", "asc").contains(sorts[1])) {
                String field = "_score";
                switch (sorts[0]) {
                    case "1": field = "hotScore";break;
                    case "2": field = "playStatNum";break;
                    case "3": field = "createTime";break;
                }

                final String fieldFinal = field;//避免变量被修改

                request.sort(s -> s
                        .field(f -> f
                                .field(fieldFinal).order(sorts[1].equals("desc") ? SortOrder.Desc : SortOrder.Asc)));
            }

        }

        request.query(boolQuery.build()._toQuery());//将BoolQuery 转换成 Query，放入 DSL 中

        //分页
        request.from((query.getPageNo() - 1) * query.getPageSize());
        request.size(query.getPageSize());

        //高亮
        request.highlight(hl -> hl
                .fields("albumTitle", hlf -> hlf
                        .preTags("<font color='red'><b><i>")
                        .postTags("</i></b></font>"))
                .fields("albumIntro", hlf -> hlf)
                .fields("announcerName", hlf -> hlf)
                .preTags("<font color='red'>")
                .postTags("</font>"));

        //排除
        request.source(ss -> ss
                .filter(f -> f
                        .excludes("hotScore", "attributeValueIndexList")));

        return request.build();
    }

    private AlbumSearchResponseVo parseResult(SearchResponse<AlbumInfoIndex> response) {
        AlbumSearchResponseVo responseVo = new AlbumSearchResponseVo();
        // 获取总记录数
        long total = response.hits().total().value();
        responseVo.setTotal(total);
        // 获取结果集
        List<AlbumInfoIndexVo> list = new ArrayList<>();
        response.hits().hits().forEach(hit -> {
            AlbumInfoIndex source = hit.source();
            AlbumInfoIndexVo albumInfoIndexVo = new AlbumInfoIndexVo();
            BeanUtils.copyProperties(source, albumInfoIndexVo);
            list.add(albumInfoIndexVo);
            // 获取高亮
            Map<String, List<String>> highlight = hit.highlight();
            if (!CollectionUtils.isEmpty(highlight)) {
                List<String> albumTitle = highlight.get("albumTitle");
                if (!CollectionUtils.isEmpty(albumTitle)) {
                    albumInfoIndexVo.setAlbumTitle(albumTitle.get(0));
                }
                List<String> albumIntro = highlight.get("albumIntro");
                if (!CollectionUtils.isEmpty(albumIntro)) {
                    albumInfoIndexVo.setAlbumIntro(albumIntro.get(0));
                }
                List<String> announcerName = highlight.get("announcerName");
                if (!CollectionUtils.isEmpty(announcerName)) {
                    albumInfoIndexVo.setAnnouncerName(announcerName.get(0));
                }
            }
        });
        responseVo.setList(list);

        return responseVo;
    }


    /**
     *  GET /suggestinfo/_search
     * {
     *   "suggest": {
     *     "titleSuggest": {
     *       "prefix": "三国",
     *       "completion": {
     *         "field": "keyword",
     *         "size": 10,    // 补全的个数
     * 			   "skip_duplicates": true // 重复跳过
     *       }
     *     },
     *     "titlePinYinSuggest": {
     *       "prefix": "sanguo",
     *       "completion": {
     *         "field": "keyword",
     *         "size": 10,    // 补全的个数
     * 			   "skip_duplicates": true // 重复跳过
     *       }
     *     },
     *     "titleSeqSuggest": {
     *       "prefix": "sg",
     *       "completion": {
     *         "field": "keyword",
     *         "size": 10,    // 补全的个数
     * 			   "skip_duplicates": true // 重复跳过
     *       }
     *     }
     *   }
     * }
     */
    @Override
    public List<String> completeSuggest(String keyword) {

        List<String> suggesters = new ArrayList<>();

        try {
            //分别针对中文前缀、拼音前缀和拼音首字母缩写前缀进行联想推荐
            SearchRequest request = SearchRequest.of(s -> s
                    .index("suggestinfo")
                    .suggest(sg -> sg
                            .suggesters("titleSuggest",ts -> ts
                                    .prefix(keyword)
                                    .completion(c->c
                                            .field("keyword")
                                            .size(10)
                                            .skipDuplicates(true)))
                            .suggesters("titlePYSuggest",ts -> ts
                                    .prefix(keyword)
                                    .completion(c->c
                                            .field("keywordPinyin")
                                            .size(10)
                                            .skipDuplicates(true)))
                            .suggesters("titleSeqSuggest",ts -> ts
                                    .prefix(keyword)
                                    .completion(c->c
                                            .field("keywordSequence")
                                            .size(10)
                                            .skipDuplicates(true)))));

            System.out.println(request);

            // 获取提示词
            SearchResponse<SuggestIndex> response = this.elasticsearchClient.search(request, SuggestIndex.class);

            Map<String, List<Suggestion<SuggestIndex>>> suggest = response.suggest();

            // 标题提示词
            List<CompletionSuggestOption<SuggestIndex>> titleOptions = suggest.get("titleSuggest").get(0).completion().options();
            // 拼音提示词
            List<CompletionSuggestOption<SuggestIndex>> titlePYOptions = suggest.get("titlePYSuggest").get(0).completion().options();
            // 首字母的提示词
            List<CompletionSuggestOption<SuggestIndex>> titleSeqOptions = suggest.get("titleSeqSuggest").get(0).completion().options();
            // 遍历放入list集合
            if (!CollectionUtils.isEmpty(titleOptions)){
                titleOptions.forEach(option -> {
                    suggesters.add(option.source().getTitle());
                });
            }
            if (!CollectionUtils.isEmpty(titlePYOptions)){
                titlePYOptions.forEach(option -> {
                    suggesters.add(option.source().getTitle());
                });
            }
            if (!CollectionUtils.isEmpty(titleSeqOptions)){
                titleSeqOptions.forEach(option -> {
                    suggesters.add(option.source().getTitle());
                });
            }

            // 判断提示词是否足够，如果不足则联想提示
            if (suggesters.size() >= 10) {
                return suggesters;
            }
            // 进行联想搜索
            SearchResponse<SuggestIndex> res = this.elasticsearchClient.search(s -> s
                    .index("suggestinfo")
                    .query(q -> q
                            .match(m -> m
                                    .field("title").query(keyword))), SuggestIndex.class);
            // 解析结果集
            List<Hit<SuggestIndex>> hitList = res.hits().hits();
            hitList.forEach(hit -> {
                if (suggesters.size() < 10){
                    suggesters.add(hit.source().getTitle());
                }
            });

            /// 去重
            /*Set<String> titleSet = new HashSet<>();
            titleSet.addAll(suggesters);
            suggesters.clear();
            suggesters.addAll(suggesters);*/
            List<String> titlelist = suggesters.stream().distinct().toList();
            return titlelist;
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    /**
     *  GET /albuminfo/_search
     * {
     *   "size": 20,
     *   "query": {
     *     "term": {
     *       "category1Id": {
     *         "value": 1
     *       }
     *     }
     *   },
     *   "sort": [
     *     {
     *       "playStatNum": {
     *         "order": "desc"
     *       }
     *     }
     *   ]
     * }
     */
    @Override
    public List<AlbumInfoIndexVo> findRankingList(Long category1Id, String statType) {
        try {
            SearchRequest request = SearchRequest.of(s -> s
                    .size(20)
                    .index("albuminfo")
                    .query(q -> q
                            .term(t -> t
                                    .field("category1Id").value(category1Id)))
                    .sort(st -> st
                            .field(f -> f
                                    .field(statType).order(SortOrder.Desc)
                            )
                    )
            );

            SearchResponse<AlbumInfoIndex> response = this.elasticsearchClient.search(request, AlbumInfoIndex.class);

            List<Hit<AlbumInfoIndex>> hitList = response.hits().hits();

            List<AlbumInfoIndexVo> albumInfoIndexVoList = hitList.stream().map(hit -> {
                AlbumInfoIndexVo albumInfoIndexVo = new AlbumInfoIndexVo();
                AlbumInfoIndex albumInfoIndex = hit.source();
                if(albumInfoIndex != null){
                    BeanUtils.copyProperties(albumInfoIndex, albumInfoIndexVo);
                }
                return albumInfoIndexVo;
                /**
                 * list.stream().collect(Collectors.toList())
                 * Collectors.toList()：返回的是一个可修改的 List（具体实现可能因 Java 版本而异，通常是 ArrayList）
                 * list.stream().toList()
                 * toList()（Java 16+ 引入）：返回的是一个不可修改的 List
                 */
            }).collect(Collectors.toList());

            return albumInfoIndexVoList;

        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    @Override
    public void syncDataToEs() {
        // 1.上一次任务的截至时间: key必须是固定的
        String startTime = (String) redisTemplate.opsForValue().get(RedisConstant.ALBUM_STAT_ENDTIME);
        if (StringUtils.isBlank(startTime)) {
            startTime = "2023-06-05 19:46:46";   // 取一个默认时间作为起始时间
        }
        // 取系统当前时间作为截至时间，并放入到redis中
        //String format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date());
        String endTime = LocalDateTime.now().toString().formatted("yyyy-MM-dd HH:mm:ss");
        this.redisTemplate.opsForValue().set(RedisConstant.ALBUM_STAT_ENDTIME, endTime);

        // 2.获取状态发生变化的专辑id集合
        Result<List<Long>> changedAlbumResult = albumInfoFeignClient.getChangedAlbumIds(startTime, endTime);
        Assert.notNull(changedAlbumResult, "同步专辑状态到ES时，获取更新状态的专辑id失败");
        List<Long> albumIds = changedAlbumResult.getData();
        Assert.notNull(albumIds, "同步专辑状态到ES时，获取更新状态的专辑id失败");

        // 可能数据量很大，分批获取变化后的数据：拆分成每个集合中有1000专辑id的小集合
        List<List<Long>> albumIdsPart = Lists.partition(albumIds, 100);
        // 为了让主线程阻塞，等待子线程执行完毕，创建一个门闩
        CountDownLatch countDownLatch = new CountDownLatch(albumIdsPart.size());
        // 3.获取变化后的状态数据，通过线程池进行优化
        albumIdsPart.forEach(subAlbumIds -> {
            executorService.execute(() -> {
                // 获取变化后的状态数据
                Result<List<AlbumStatVo>> statsByAlbumResult = albumInfoFeignClient.getAlbumStatsByAlbumIds(subAlbumIds);
                Assert.notNull(statsByAlbumResult, "同步专辑状态到ES时，远程调用根据id获取状态失败");
                List<AlbumStatVo> albumStatVos = statsByAlbumResult.getData();
                Assert.notNull(albumStatVos, "同步专辑状态到ES时，远程调用根据id获取状态失败");
                // 同步到ES
                albumStatVos.forEach(albumStatVo -> {
                    AlbumInfoIndex albumInfoIndex = new AlbumInfoIndex();
                    BeanUtils.copyProperties(albumStatVo, albumInfoIndex);
                    albumInfoIndex.setId(albumStatVo.getAlbumId());
                    albumInfoIndex.setHotScore(albumStatVo.getPlayStatNum() * 0.1
                            + albumStatVo.getSubscribeStatNum() * 0.3
                            + albumStatVo.getBuyStatNum() * 0.4
                            + albumStatVo.getCommentStatNum() * 0.2
                    );
                    this.elasticsearchTemplate.update(albumInfoIndex);
                });
                countDownLatch.countDown();
            });
        });
        try {
            countDownLatch.await();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}
